package shardmaster

import (
	crand "crypto/rand"
	"encoding/gob"
	"fmt"
	"log"
	"math/big"
	"math/rand"
	"net"
	"net/rpc"
	"os"
	"paxos"
	"sync"
	"sync/atomic"
	"syscall"
)

type ShardMaster struct {
	mu         sync.Mutex
	l          net.Listener
	me         int
	dead       int32 // for testing
	unreliable int32 // for testing
	px         *paxos.Paxos

	configs []Config // indexed by config num
}

type Type string

const (
	JOIN  Type = "join"
	LEAVE Type = "leave"
	MOVE  Type = "move"
	QUERY Type = "query"
)

type Op struct {
	// Your data here.
	T       Type

	GID     int64
	Servers []string
	Shard   int

	Seq     int64
}

func nrand() int64 {
	bigx, _ := crand.Int(crand.Reader, big.NewInt(int64(1)<<62))
	return bigx.Int64()
}

func (sm *ShardMaster) Join(args *JoinArgs, reply *JoinReply) error {
	// Your code here.
	sm.mu.Lock()
	defer sm.mu.Unlock()
	op := Op{
		Seq:     nrand(),
		T:       JOIN,
		GID:     args.GID,
		Servers: args.Servers}
	sm.px.MakePropose(op)
	return nil
}

func (sm *ShardMaster) Leave(args *LeaveArgs, reply *LeaveReply) error {
	// Your code here.

	sm.mu.Lock()
	defer sm.mu.Unlock()

	sm.px.MakePropose(Op{Seq: nrand(), T: LEAVE, GID: args.GID})
	return nil
}

func (sm *ShardMaster) Move(args *MoveArgs, reply *MoveReply) error {
	// Your code here.
	sm.mu.Lock()
	defer sm.mu.Unlock()

	sm.px.MakePropose(Op{Seq: nrand(), T: MOVE, GID: args.GID, Shard: args.Shard})
	return nil
}

func (sm *ShardMaster) Query(args *QueryArgs, reply *QueryReply) error {
	// Your code here.
	sm.mu.Lock()
	defer sm.mu.Unlock()

	sm.px.MakePropose(Op{Seq: nrand(), T: QUERY})

	if args.Num < 0 || args.Num >= len(sm.configs) {
		args.Num = len(sm.configs) - 1
	}

	reply.Config = sm.configs[args.Num].Clone()
	return nil
}

func checkOp(decidedOp, originOp interface{}) bool {
	return decidedOp.(Op).Seq == originOp.(Op).Seq
}

func (sm *ShardMaster) handle(decidedOp interface{}) {
	decided := decidedOp.(Op)

	if decided.T == QUERY {
		return
	}
	newConfig := sm.configs[len(sm.configs)-1].Clone()
	switch decided.T {
	case JOIN:
		newConfig.Groups[decided.GID] = decided.Servers

		if len(newConfig.Groups) <= NShards {
			newConfig.shard()
		}
		//fmt.Println(sm.me, "join", decided.GID, decided.Servers, newConfig.Shards)
	case LEAVE:
		delete(newConfig.Groups, decided.GID)
		for _, v := range newConfig.Shards {
			if v == decided.GID {
				newConfig.shard()
				break
			}
		}
		//fmt.Println(sm.me, "leave", decided.GID, newConfig.Shards)
	case MOVE:
		newConfig.Shards[decided.Shard] = decided.GID
		//fmt.Println(sm.me, "move", decided.Shard, decided.GID)
	}
	newConfig.Num = len(sm.configs)
	sm.configs = append(sm.configs, newConfig)
	return
}

func (c *Config) Clone() Config {
	newC := Config{}
	newC.Num = c.Num
	newC.Shards = [NShards]int64{}
	for k, v := range c.Shards {
		newC.Shards[k] = v
	}
	newC.Groups = make(map[int64][]string)
	for k, v := range c.Groups {
		newC.Groups[k] = make([]string, len(v))
		for k1, v1 := range v {
			newC.Groups[k][k1] = v1
		}
	}
	return newC
}

func (c *Config) GetGidSlice(shardSlice []int, me int64) []int64 {

	result := make([]int64, 0)

	for _, v := range shardSlice {
		gid := c.Shards[v]
		in := false
		for _, v2 := range result {
			if v2 == gid {
				in = true
				break
			}
		}
		if !in && gid != me && gid != 0 {
			result = append(result, gid)
		}
	}
	return result
}

func (c *Config) GetShard(gid int64) []int {

	result := make([]int, 0)

	for k, v := range c.Shards {
		if v == gid {
			result = append(result, k)
		}
	}
	return result
}

func (c *Config) shard() {
	groups := make([]int64, len(c.Groups))
	i := 0
	for k, _ := range c.Groups {
		groups[i] = k
		i++
	}
	for i := 0; i < NShards; i++ {
		c.Shards[i] = groups[(i+1)%len(groups)]
	}
}

// please don't change these two functions.
func (sm *ShardMaster) Kill() {
	atomic.StoreInt32(&sm.dead, 1)
	sm.l.Close()
	sm.px.Kill()
}

// call this to find out if the server is dead.
func (sm *ShardMaster) isdead() bool {
	return atomic.LoadInt32(&sm.dead) != 0
}

// please do not change these two functions.
func (sm *ShardMaster) setunreliable(what bool) {
	if what {
		atomic.StoreInt32(&sm.unreliable, 1)
	} else {
		atomic.StoreInt32(&sm.unreliable, 0)
	}
}

func (sm *ShardMaster) isunreliable() bool {
	return atomic.LoadInt32(&sm.unreliable) != 0
}

//
// servers[] contains the ports of the set of
// servers that will cooperate via Paxos to
// form the fault-tolerant shardmaster service.
// me is the index of the current server in servers[].
//
func StartServer(servers []string, me int) *ShardMaster {
	sm := new(ShardMaster)
	sm.me = me

	sm.configs = make([]Config, 1)
	sm.configs[0].Groups = map[int64][]string{}

	rpcs := rpc.NewServer()

	gob.Register(Op{})
	rpcs.Register(sm)
	sm.px = paxos.Make(servers, me, rpcs)
	sm.mu = sync.Mutex{}

	sm.px.SetFunc(checkOp, sm.handle)

	os.Remove(servers[me])
	l, e := net.Listen("unix", servers[me])
	if e != nil {
		log.Fatal("listen error: ", e)
	}
	sm.l = l

	// please do not change any of the following code,
	// or do anything to subvert it.

	go func() {
		for sm.isdead() == false {
			conn, err := sm.l.Accept()
			if err == nil && sm.isdead() == false {
				if sm.isunreliable() && (rand.Int63()%1000) < 100 {
					// discard the request.
					conn.Close()
				} else if sm.isunreliable() && (rand.Int63()%1000) < 200 {
					// process the request but force discard of reply.
					c1 := conn.(*net.UnixConn)
					f, _ := c1.File()
					err := syscall.Shutdown(int(f.Fd()), syscall.SHUT_WR)
					if err != nil {
						fmt.Printf("shutdown: %v\n", err)
					}
					go rpcs.ServeConn(conn)
				} else {
					go rpcs.ServeConn(conn)
				}
			} else if err == nil {
				conn.Close()
			}
			if err != nil && sm.isdead() == false {
				fmt.Printf("ShardMaster(%v) accept: %v\n", me, err.Error())
				sm.Kill()
			}
		}
	}()

	return sm
}
